package com.lab.hbase.labAPI;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.ColumnPrefixFilter;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.filter.FilterList.Operator;
import org.apache.hadoop.hbase.filter.ColumnRangeFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

/**
 * hbase 过滤器 测试类
 * 
 * @author hadoop
 *
 */
public class HbaseFilterTest {

	public static void main(String[] args) throws Exception {
		// TODO Auto-generated method stub
		HbaseFilterTest hft = new HbaseFilterTest();
		// hbase 初始化连接
		hft.init();
		//查询单条数据
		hft.querySingleData();
		System.out.println("");
		// 单列值过滤器
		hft.scanDataBySingleColumnValueFilter();
		System.out.println("");
		// rowkey过滤器
		hft.scanDataByRowFilter();
		System.out.println("");
		// 列前缀过滤器
		hft.scanDataByColumnPrefixFilter();
		System.out.println("");
		
//过滤器集合
		hft.scanDataByFilterList();
		System.out.println("");

		// 分页过滤器
		hft.scanDataByPageFilter();
		System.out.println("");

		// 列前缀过滤器2
		hft.testColumnPrefixFilter();
		System.out.println("");

		// 行过滤器2
		hft.testRowFilter();
		System.out.println("");

		// 关闭连接
		hft.close();

	}

	/**
	 * 配置ss
	 */
	static Configuration config = null;
	private Connection connection = null;
	private Table table = null;

	// private Admin admin = null;
	@Before
	public void init() throws Exception {
		System.out.println("#########连接初始化");
		// 一、 创建hbase配置信息，默认读取hbase-site.xml文件内容进行配置
		config = HBaseConfiguration.create();
		System.out.println(config);
		// 修改该配置信息
		// config.set("hbase.zookeeper.quorum", "slave1,slave2,slave3");// zookeeper地址
		// config.set("hbase.zookeeper.property.clientPort", "2181");// zookeeper端口

		// 二、获取hbase连接
		connection = ConnectionFactory.createConnection(config);
		System.out.println("连接是否关闭： " + connection.isClosed());

		// 获取待处理的table对象
		table = connection.getTable(TableName.valueOf("t_user"));
		System.out.println("表：" + table);
		System.out.println("#########连接初始化完成");
	}

	@After
	public void close() throws Exception {
		System.out.println("close table and connection on hbase.");
		table.close();

		connection.close();
	}

	/**
	 * 单条查询
	 * 
	 * @throws Exception
	 */
	@Test
	public void querySingleData() throws Exception {
		System.out.println("querySingleData");
		Get get1 = new Get("rowkey_wang_21".getBytes());
		Result result = table.get(get1);
		System.out.print("rowkey：" + Bytes.toString(result.getRow()));
		System.out.print(
				" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
		System.out.println(
				"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));

	}

	/**
	 * 全表扫描的过滤器 列值过滤器
	 * 
	 * @throws Exception
	 */
	@Test
	public void scanDataBySingleColumnValueFilter() throws Exception {
		System.out.println("scanDataBySingleColumnValueFilter");
		// 创建全表扫描的scan
		Scan scan = new Scan();
		// 过滤器：列值过滤器
		SingleColumnValueFilter filter = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
				CompareFilter.CompareOp.EQUAL, Bytes.toBytes("zhangsan2"));
		// 设置过滤器
		scan.setFilter(filter);

		// 打印结果集
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));

		}

	}

	/**
	 * rowkey过滤器
	 * 
	 * @throws Exception
	 */
	@Test
	public void scanDataByRowFilter() throws Exception {
		System.out.println("scanDataByRowFilter");
		// 创建全表扫描的scan
		Scan scan = new Scan();
		// 匹配rowkey
		RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("^rowkey_wang"));
		// 设置过滤器
		scan.setFilter(filter);
		// 打印结果集
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));

		}

	}

	/**
	 * 列名前缀过滤器
	 * 
	 * @throws Exception
	 */
	@Test
	public void scanDataByColumnPrefixFilter() throws Exception {
		System.out.println("scanDataByColumnPrefixFilter");
		// 创建全表扫描的scan
		Scan scan = new Scan();
		// 匹配ColumnPrefix
		ColumnPrefixFilter filter = new ColumnPrefixFilter(Bytes.toBytes("name2"));
		// 设置过滤器
		scan.setFilter(filter);
		// 打印结果集
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.print(" name222:"
					+ Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name222"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));

		}

	}

	/**
	 * 过滤器集合
	 * 
	 * @throws Exception
	 */
	@Test
	public void scanDataByFilterList() throws Exception {
		System.out.println("scanDataByFilterList");
		// 创建全表扫描的scan
		Scan scan = new Scan();
		// 过滤器集合：MUST_PASS_ALL（and）,MUST_PASS_ONE(or)
		FilterList filterList = new FilterList(Operator.MUST_PASS_ONE);
		// 匹配rowkey
		RowFilter filter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator("^rowkey_zhang"));
		// 匹配name的值等于wangsenfeng
		SingleColumnValueFilter filter2 = new SingleColumnValueFilter(Bytes.toBytes("info"), Bytes.toBytes("name"),
				CompareFilter.CompareOp.EQUAL, Bytes.toBytes("zhangsan"));
		filterList.addFilter(filter);
		filterList.addFilter(filter2);
		// 设置过滤器
		scan.setFilter(filterList);
		// 打印结果集
		ResultScanner scanner = table.getScanner(scan);
		for (Result result : scanner) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));

		}

	}

	/*
	 * 测试列过滤器
	 */
	public void testColumnPrefixFilter() throws IOException {

		System.out.println("#########列过滤器##################################################");

		// 三、创建Scan对象
		Scan scan = new Scan();

		// 四-1、查询列前缀为"nam" 或 值包含“zhangsan”的行
		System.out.println("列前缀为nam 或 值包含zhangsan的行");
		// 设置过滤器。
		Filter filter1 = new ColumnPrefixFilter("nam".getBytes());
		Filter filter2 = new SingleColumnValueFilter("info".getBytes(), "name".getBytes(), CompareOperator.EQUAL,
				new SubstringComparator("zhangsan"));

		FilterList fl = new FilterList(FilterList.Operator.MUST_PASS_ONE);
		fl.addFilter(filter1);
		fl.addFilter(filter2);

		// 将过滤器设置给Scan对象
		scan.setFilter(fl);
		// 获得查询结果
		ResultScanner scanner1 = table.getScanner(scan);
		// 遍历结果
		for (Result result : scanner1) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));
		}
		// 释放ResultScanner资源
		scanner1.close();
	}

	/*
	 * 测试行过滤器
	 */
	public void testRowFilter() throws IOException {
		System.out.println("#########行过滤器##################################################");

		// 三、创建Scan对象
		Scan scan = new Scan();

		// 四-1、查询rowkey小于等于rowkey_li_12的行
		System.out.println("rowkey小于等于rowkey_li_12的行");
		// 设置过滤器。第一个参数为CompareOp，表示小于或等于；第二个参数为行健值。
		Filter filter1 = new RowFilter(CompareFilter.CompareOp.LESS_OR_EQUAL, new BinaryComparator("rowkey_li_12".getBytes()));
		// 将过滤器设置给Scan对象
		scan.setFilter(filter1);
		// 获得查询结果
		ResultScanner scanner1 = table.getScanner(scan);
		// 遍历结果
		for (Result result : scanner1) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));
		}
		// 释放ResultScanner资源
		scanner1.close();

		// 四-2、正则获取rowkey结尾为2的行
		System.out.println("正则获取rowkey结尾为2的行");
		// 设置过滤器。第一个参数为CompareOp，表示等于；第二个参数为正则表达式，表示结尾含有"2"。
		Filter filter2 = new RowFilter(CompareOperator.EQUAL, new RegexStringComparator(".*2$"));
		// 将过滤器设置给Scan对象
		scan.setFilter(filter2);
		// 获得查询结果
		ResultScanner scanner2 = table.getScanner(scan);
		// 遍历结果
		for (Result result : scanner2) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));
		}
		// 释放ResultScanner资源
		scanner2.close();

		// 四-3、获取rowkey包含有1的行
		System.out.println("rowkey包含有9的行");
		// 设置过滤器。第一个参数为CompareOp，表示等于；第二个参数SubstringComparator，表示字符串中含有"9"。
		Filter filter3 = new RowFilter(CompareOperator.EQUAL, new SubstringComparator("9"));
		// 将过滤器设置给Scan对象
		scan.setFilter(filter3);
		// 获得查询结果
		ResultScanner scanner3 = table.getScanner(scan);
		// 遍历结果
		for (Result result : scanner3) {
			System.out.print("rowkey：" + Bytes.toString(result.getRow()));
			System.out.print(
					" name:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
			System.out.println(
					"password:" + Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));
		}
		// 释放ResultScanner资源
		scanner3.close();

	}

	/*
	 * 分页过滤器
	 */
	public void scanDataByPageFilter() throws IOException {

		System.out.println("#########分页过滤器##################################################");

		// 三、创建Scan对象
		// 1、创建过滤器PageFilter。该过滤器表示按行分页。参数3表示每个分页有3行记录。
		Filter filter = new PageFilter(3);
		// POSTFIX=0
		final byte[] POSTFIX = new byte[] { 0x00 };
		int totalRows = 0;
		byte[] lastRow = null;
		// 2、进入循环。为了演示效果，这里遍历所有符合条件的数据，需要循环输出。
		while (true) {
			// 3、初始化Scan实例。该实例用于查询符合条件的数据。
			Scan scan = new Scan();
			// 4、设置过滤器。将前面创建好的分页过滤器设置到Scan实例中。
			scan.setFilter(filter);
			// 5、设置遍历的开始位置。即表示开始的行健位置，如果是第一次循环（即第一页），则不进入该语句块。
			if (lastRow != null) {
				// 注意这里添加了POSTFIX操作，不然死循环了
				byte[] startRow = Bytes.add(lastRow, POSTFIX);
				System.out.println("start row:" + Bytes.toStringBinary(startRow));//
				scan.setStartRow(startRow);
			}
			// 6、执行查询。使用HTable实例执行扫描查询，并且将扫描结果输出，并且给行健遍历赋值。
			ResultScanner scanner = table.getScanner(scan);
			int localRows = 0;
			Result result;

			// 输出一页的结果。
			while ((result = scanner.next()) != null) {
				System.out.print("rowkey：" + Bytes.toString(result.getRow()));
				System.out.print(" name:"
						+ Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("name"))) + " | ");
				System.out.println("password:"
						+ Bytes.toString(result.getValue(Bytes.toBytes("info"), Bytes.toBytes("password"))));

				totalRows++;
				localRows++;//
				lastRow = result.getRow();
			}
			System.out.println("");
			// 7、关闭ResultScanner实例。
			scanner.close();
			// 8、跳出循环条件
			if (localRows == 0)
				break;
		}
		System.out.println("total rows:" + totalRows);

	}

}
